package com.lxj.app.dwd;


import com.lxj.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

public class DwdUserRegister {
    /**
     * 用户域用户注册事实表应用
     * 负责处理用户注册事件数据，记录用户注册的详细信息
     * 主要功能点：
     * - 从Kafka读取用户注册相关数据
     * - 解析用户注册事件信息
     * - 提取用户注册时间、注册渠道、设备信息等
     * - 将处理后的数据写入Kafka主题
     * 
     * 数据流向：
     * 用户注册事件数据 -> DwdUserRegister -> 处理后写入dwd_user_register主题
     */
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备
    // 创建流执行环境，设置并行度为1（开发测试环境）
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    // 创建表执行环境，设置时区为GMT+8（北京时间）
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

    // TODO 2. 启用状态后端

    // 预留位置，用于配置状态后端（如RocksDB）


    // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
    // 创建Kafka源表topic_db，消费业务数据库变更日志
    tableEnv.executeSql("create table topic_db(" +
            "`database` string, " +  // 数据库名
            "`table` string, " +     // 表名
            "`type` string, " +      // 操作类型(insert/update/delete)
            "`data` map<string, string>, " +  // 变更数据内容
            "`ts` string " +         // 变更时间戳
            ")" + MyKafkaUtil.getKafkaDDL("topic_db", "dwd_trade_order_detail_211126"));

    // TODO 4. 读取用户表数据
    // 从topic_db中过滤出user_info表的insert操作数据
    Table userInfo = tableEnv.sqlQuery("select " +
            "data['id'] user_id, " +       // 用户ID
            "data['create_time'] create_time, " +  // 用户创建时间
            "ts " +                       // 时间戳
            "from topic_db " +
            "where `table` = 'user_info' " +  // 只处理user_info表
            "and `type` = 'insert' ");    // 只处理新增数据
    tableEnv.createTemporaryView("user_info", userInfo);  // 注册为临时视图供后续使用

    // TODO 5. 创建 Kafka-Connector dwd_user_register 表
    // 创建Kafka目标表，用于存储用户注册事实数据
    tableEnv.executeSql("create table `dwd_user_register`( " +
            "`user_id` string, " +       // 用户ID
            "`date_id` string, " +       // 日期分区(yyyy-MM-dd格式)
            "`create_time` string, " +   // 创建时间
            "`ts` string " +             // 时间戳
            ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_user_register"));
    // TODO 6. 将输入写入 Kafka-Connector 表
    // 将用户注册数据写入目标Kafka主题
    tableEnv.executeSql("insert into dwd_user_register " +
            "select  " +
            "user_id, " +
            "date_format(create_time, 'yyyy-MM-dd') date_id, " +  // 将时间格式化为日期
            "create_time, " +
            "ts " +
            "from user_info");
    // 开发调试用：打印查询结果
    tableEnv.sqlQuery("select  *  from  user_info").execute().print();

    // 执行作业
//    env.execute();
}
}
